package ML

import org.apache.spark.rdd.RDD

/**
  * Created by THINKPAD on 2017/8/7.
  * 使用DBSCAN的方法，给职住锚点聚类，划分城市空间
  */
object DBSCAN_anchor {
  def main(args: Array[String]) {

  }

  def anchor_density(data:RDD[String], homeIndex:Int, workIndex:Int, minPts:Int, sep:String = "\t", dis:Double):String = {
    //    val homeArr = data.map(_.split(sep)(homeIndex)).collect()
    //    val workArr = data.map(_.split(sep)(workIndex)).collect()
    //    homeArr.map(f=>(three(f.split(",")(0).toDouble ), three( f.split(",")(1).toDouble)))

    //    homeArr.map(f=>(f.split(",")(0).toDouble, f.split(",")(1).toDouble ))
    val homeArr = data.map(_.split(sep)(homeIndex)).map(f=>(f.split(",")(0).toDouble ,  f.split(",")(1).toDouble))
    "whatever"
  }
  def three (a:Double) = {
    BigDecimal(a).setScale(3, BigDecimal.RoundingMode.HALF_UP).toDouble
  }

  def C(LatA: Double, LonA: Double, LatB: Double, LonB: Double) = Math.sin(LatA / 57.2958) * Math.sin(LatB / 57.2958) + Math.cos(LatA / 57.2958) * Math.cos(LatB / 57.2958) * Math.cos((LonA - LonB) / 57.2958)
  def dis(lona:String, lata:String,lonb:String,latb:String) = {
    val LatA = lata.toDouble
    val LatB = latb.toDouble
    val LonA = lona.toDouble
    val LonB = lonb.toDouble
    if (LatA*LatB*LonA*LonB != 0) {
      val dis =  (6371004 * Math.acos(C(LatA, LonA, LatB, LonB)) * 1000).toInt / 1000.toDouble
      BigDecimal(dis).setScale(3, BigDecimal.RoundingMode.HALF_UP).toDouble
    }
    else {"nodata"}
  }
}
